package net.quanter.shield.mq.rocketmq.consumer;

import lombok.extern.slf4j.Slf4j;
import net.quanter.shield.mq.MQConsumer;
import net.quanter.shield.mq.RunType;
import net.quanter.shield.mq.rocketmq.enums.RocketMqType;
import net.quanter.shield.mq.rocketmq.param.RocketMQBorkerParam;

@Slf4j
public class DefaultMQPushConsumerWrapper {
    MQConsumer rocketMQConsumer;
    RunType runType;

    public DefaultMQPushConsumerWrapper(
            RocketMqType type,
            RunType runType,
            RocketMQBorkerParam mqConnectVO,
            String groupId,
            RocketMQConsumerParam... mqConsumerConnectVOS
    ) throws Exception {
        if (type == null || mqConnectVO == null || groupId == null || mqConsumerConnectVOS == null) {
            throw new Exception("DefaultMQPushConsumerWrapper constuct error,some params is null");
        }
        this.runType = runType;
        if (type == RocketMqType.TCP) {
            rocketMQConsumer = new RocketMQTcpConsumer(mqConnectVO,groupId,mqConsumerConnectVOS);
        } else if(type==RocketMqType.HTTP){
            rocketMQConsumer = new RocketMQHttpConsumner(mqConnectVO, groupId,mqConsumerConnectVOS);
        }else {
            rocketMQConsumer = new RocketMQCommunityConsumer(mqConnectVO,groupId,mqConsumerConnectVOS);
        }
    }

    public void start() throws Exception {
        rocketMQConsumer.start(runType);
    }

    public void stop() {
        rocketMQConsumer.stop();
    }

}